package com.bigdata

import java.lang
import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.Date

import com.bigdata.bean.{LogEvent, UrlViewCount}
import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

object TrafficAnalysis {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val fileSource: DataStream[String] = env.readTextFile(".\\NetworkTrafficAnalysis\\src\\main\\resources")

    fileSource.map {
      log =>
        val items: Array[String] = log.split(" ")
        val simpleDateFormat = new SimpleDateFormat("MM/dd/yyyy:HH:mm:ss")
        val date: Date = simpleDateFormat.parse(items(3))
        val timestamp: Long = date.getTime
        LogEvent(items(0), items(2), timestamp, items(5), items(6))
    }
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LogEvent](Time.milliseconds(1000)) {
        override def extractTimestamp(element: LogEvent): Long = element.eventTime
      })
      .filter(_.method == "GET")
      .keyBy(_.url)
      .timeWindow(Time.minutes(10), Time.seconds(5))
      .aggregate(new CountAgg, new WindowResultFunction)
      .keyBy(_.windowEnd)
      .process(new TopNHotUrl(5))
      .writeAsText("./output/hoturls.txt")
      .setParallelism(1)

    env.execute("traffic analysis jod")

  }

  class CountAgg extends AggregateFunction[LogEvent, Long, Long] {
    override def createAccumulator(): Long = 0L

    override def add(value: LogEvent, accumulator: Long): Long = accumulator + 1L

    override def getResult(accumulator: Long): Long = accumulator

    override def merge(a: Long, b: Long): Long = a + b
  }

  class WindowResultFunction extends WindowFunction[Long, UrlViewCount, String, TimeWindow] {
    override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[UrlViewCount]): Unit = {
      out.collect(UrlViewCount(key, window.getEnd, input.iterator.next()))
    }
  }

  class TopNHotUrl(n: Int) extends KeyedProcessFunction[Long, UrlViewCount, String]{

    lazy val listState: ListState[UrlViewCount] = getRuntimeContext.getListState[UrlViewCount](new ListStateDescriptor[UrlViewCount]("list_state_desc", classOf[UrlViewCount]))
    override def processElement(value: UrlViewCount, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#Context, out: Collector[String]): Unit = {
        listState.add(value)
        ctx.timerService().registerEventTimeTimer(value.windowEnd+1)
    }

    override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, UrlViewCount, String]#OnTimerContext, out: Collector[String]): Unit = {

      import scala.collection.JavaConversions._
      val countsToViewCounts: ListBuffer[UrlViewCount] = ListBuffer()

      for(item <- listState.get()){
        countsToViewCounts.add(item)
      }

      listState.clear()


      // 按照点击量从大到小排序
      val sortedItems: ListBuffer[UrlViewCount] = countsToViewCounts.sortBy(_.count)(Ordering.Long.reverse).take(n)

      // 将排名信息格式化成 String, 便于打印
      val result: StringBuilder = new StringBuilder
      result.append("====================================\n")
      result.append(s"时间: ${new Timestamp(timestamp - 1)}\n")

      for (i <- sortedItems.indices) {
        val currentItem: UrlViewCount = sortedItems(i)
        // e.g.  No1：  url=12224  访问量=2413
        result.append(s"No.${i+1}:\tUrl=${currentItem.url}\t访问量=${currentItem.count}\n")

      }
      result.append("====================================\n\n")
      // 控制输出频率，模拟实时滚动结果
      //      Thread.sleep(1000)
      out.collect(result.toString)

    }
  }

}
